package com.ibm.cps.redis;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import com.google.common.base.Throwables;
import com.ibm.util.ErrorCode;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;

public class RuleStateRedis implements KryoSerializable {

	private String redisHost;
	private int redisPort;
	private JedisPool pool;
	private static RuleStateRedis ruleStateRedis = null;
	private static Logger logger;

	public static RuleStateRedis getRuleStateRedis(String redisHost,
			int redisPort) {
		logger = LoggerFactory.getLogger(RuleStateRedis.class);
		if (ruleStateRedis == null)
			synchronized (RuleStateRedis.class) {
				if (ruleStateRedis == null) {
					ruleStateRedis = new RuleStateRedis(redisHost, redisPort);
				}
			}
		return ruleStateRedis;
	}

	private RuleStateRedis(String redisHost, int redisPort) {
		this.redisHost = redisHost;
		this.redisPort = redisPort;
		JedisPoolConfig configure = new JedisPoolConfig();
		configure.setMaxTotal(LocalConfig.REDIS_MAX_CONNECTIONS_NUM);
		configure.setMaxWaitMillis(1000 * LocalConfig.REDIS_WAITING_SECONDS);
		logger.info("Attempting to connect redis " + redisHost + ":" + redisPort);
		pool = new JedisPool(configure, redisHost, redisPort);
	}
	
	@Override
	public void read(Kryo kryo, Input input) {
		redisHost = input.readString();
		redisPort = input.readInt();
		JedisPoolConfig configure = new JedisPoolConfig();
		configure.setMaxTotal(LocalConfig.REDIS_MAX_CONNECTIONS_NUM);
		configure.setMaxWaitMillis(1000 * LocalConfig.REDIS_WAITING_SECONDS);
		logger.info("Attempting to connect redis " + redisHost + ":" + redisPort);
		logger.info("===== RuleStateRedis: opening redis connection on read() =====");
		pool = new JedisPool(configure, redisHost, redisPort);
	}

	@Override
	public void write(Kryo kryo, Output output) {
		output.writeString(redisHost);
		output.writeInt(redisPort);
		logger.info("===== RuleStateRedis: closing redis connection on read() =====");
		pool.close();
	}
	
	public String getRedisHost() {
		return redisHost;
	}

	public void setRedisHost(String redisHost) {
		this.redisHost = redisHost;
	}

	public int getRedisPort() {
		return redisPort;
	}

	public void setRedisPort(int redisPort) {
		this.redisPort = redisPort;
	}

	public HashMap<String, Boolean> readRuleState(String rule)
			throws CPSException {
		Jedis connection = null;
		try {
			HashMap<String, Boolean> ds2State = new HashMap<String, Boolean>();
			byte[] id = rule.toString().getBytes();
			int numActives = pool.getNumActive();
			logger.info("Redis active threads before get: " + numActives);
			connection = pool.getResource();

			numActives = pool.getNumActive();
			logger.info("Redis active threads after get: " + numActives);

			Set<String> keys = connection.keys("*");
			if (!keys.contains(rule.toString())) {
				logger.info("This is a new rule.");
				return null;
			}
			Map<byte[], byte[]> result = connection.hgetAll(id);

			String ds_ts;
			String previousState;
			if (result == null || result.size() < 1) {
				logger.info("No message in this rule.");
				return null;
			}
			for (Map.Entry<byte[], byte[]> entry : result.entrySet()) {
				ds_ts = new String(entry.getKey());
				previousState = new String(entry.getValue());
				if (previousState.equals("1")) {
					ds2State.put(ds_ts, true);
				} else {
					ds2State.put(ds_ts, false);
				}
			}
			numActives = pool.getNumActive();
			logger.info("Redis active threads after connection close: "
					+ numActives);
			return ds2State;
		} catch (Exception e) {
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"Can't read rule state from redis successfully.");
		} finally {
			if (connection != null) {
				connection.close();
			}
		}

	}

	public void updateRuleState(String rule, String key, boolean state)
			throws CPSException {
		Jedis connection = null;
		try {
			connection = pool.getResource();
			connection.hset(rule.toString().getBytes(), key.getBytes(),
					(state ? "1" : "0").getBytes());
		} catch (Exception e) {
			logger.error(Throwables.getStackTraceAsString(e));
			// throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
			// "Can't get the resource from the pool within "
			// + LocalConfig.REDIS_WAITING_SECONDS
			// + " seconds or can't set key value");
		} finally {
			if (connection != null) {
				connection.close();
			}
		}
	}

	public static void main(String[] args) {
	}


}
